package com.chencong.online.utils;

import com.chencong.online.common.ConfigCommon;
import com.starrocks.connector.flink.StarRocksSink;
import com.starrocks.connector.flink.row.StarRocksSinkRowBuilder;
import com.starrocks.connector.flink.table.StarRocksSinkOptions;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.TableSchema;

/**
 * @program: bigdata
 * @ClassName StarRocksUtil
 * @description:StartRocks工具类
 * @author: chencong
 * @create: 2021-12-17 18:06
 **/
public class StarRocksUtil {
    /**
     * @return
     * @Author chencong
     * @Description sink StarRocks函数
     * @Date 16:39 2021/12/29
     * @Param
     **/
    public static <T> SinkFunction<T> sinkStarRocks(TableSchema tableSchema, String tableName, String databaseName,
                                                    StarRocksSinkRowBuilder<T> rowDataTransformer) {
        SinkFunction<T> StarRocksSinkBuilder = StarRocksSink.sink(
                //1、 the table structure
                tableSchema,
                //2、 the sink options
                StarRocksSinkOptions.builder()
//                        .withProperty("jdbc-url", "jdbc:mysql://192.168.5.40:9030?characterEncoding=utf-8&useSSL=false") //query_port
                        .withProperty("jdbc-url", "jdbc:mysql://" + ConfigCommon.TEST40_STARROCKS_HOST + ":" + ConfigCommon.STARROCKS_QUERY_PORT + "?characterEncoding=utf-8&useSSL=false") //query_port
//                        .withProperty("load-url", "192.168.5.40:8030")//http_port
                        .withProperty("load-url", ConfigCommon.TEST40_STARROCKS_HOST + ":" + ConfigCommon.STARROCKS_HTTP_PORT)//http_port
                        .withProperty("username", "root")
                        .withProperty("password", "")
                        .withProperty("table-name", tableName)
                        .withProperty("database-name", databaseName)
                        //设置列分隔符
                        .withProperty("sink.properties.column_separator", "\\x01")
                        //设置行分隔符
                        .withProperty("sink.properties.row_delimiter", "\\x02")
                        //设置sink提交周期，这里设置10s提交一次
                        .withProperty("sink.buffer-flush.interval-ms", "10000")
                        .build(),
                //3、set the slots with streamRowData
                rowDataTransformer
        );
        return StarRocksSinkBuilder;
    }

}
